#include "StdAfx.h"
#include "PortRouter.h"
#include "PortRouterMessage.h"
#include "DynamicBindingManager.h"
#include "LocalEndPoint.h"
#include "ContractChecker.h"
#include "ProtocolFactory.h"
#include "RemoteReplyEndpoint.h"

#include <RoomFW.h>		// Brings in G_LOG macros.

#ifdef _DEBUG
#define DBGMSG(msg) {printf##msg; fflush(stdin);}
#else
#define DBGMSG(msg)
#endif

namespace RoomFW {

// Make sure if you enable G_TRACE_METHOD, that you've initialized the logger in your setup!
#if 1		// DEBUG -- hansohn, 15 May 2009
    #define TRACE_ENTRY(m)
// 	#define TRACE_ENTRY(m) G_TRACE_METHOD(m)
	#define SS_TRACE0 G_LOG_0_TRACE
	#define SS_TRACE1 G_LOG_1_TRACE
#else
    #define TRACE_ENTRY(m)
	#define SS_TRACE0
	#define SS_TRACE1
#endif

/******************
 * Public Methods *
 ******************/
PortRouter& PortRouter::Instance(bool reset)
{
	static PortRouter *pPR=NULL;
	if( NULL == pPR ) {
		pPR = new PortRouter ;
	}else if( reset ){
		delete pPR;
		pPR = new PortRouter ;
	}
	return *pPR;
}


PortRouter::PortRouter() :
	nextPortIndex_(0),
	localPort_(0),
	localAddress_(),
	localEndPoints_(0),
	remoteEndPoints_(0)
{
	bindAckEvent_.create();
	bindAckWait_.create(true);
}


PortRouter::~PortRouter()
{
	// Remove all bindings.
	remoteRegistrations_.clear();
	remoteBindings_.clear();
	localEndPoints_.clearAll();
	remoteEndPoints_.clearAll();

	// Unregister with DBM.
	DynamicBindingManager::Instance().unregister(this);
}

bool PortRouter::start(int port, const char* address, uint32_t timeout)
{
	TRACE_ENTRY(PortRouter::start);

	localPort_ = port;
	localAddress_ = address;
	return __super::start(port, address, timeout);
}

SimpleSocket::SocketPtr PortRouter::connect(int port, const char* address, unsigned attempts, unsigned attemptIntervalMsec)
{
	TRACE_ENTRY(PortRouter::connect);

	SimpleSocket::SocketPtr client;
	for (int attempt = 0; attempt < attempts; attempt++)
	{
		client = __super::connect(port, address);
		if (client)
			break;

		G_LOG_4_WARNING("PortRouter.connect(%d, %s) attempt %d failed.  Sleeping %d msec.", port, address, attempt, attemptIntervalMsec);
		Sleep(attemptIntervalMsec);
	}

	if (client)
	{
		CriticalSectionGuard g(cs_);
		clients_.insert(client);
	}
	return client;
}

void PortRouter::registerPort(const char* portName, IContractCheckerSPC contract)
{
	// This is called by DBM when a local port is registered.
	//
	// We first of all save the name and contract info, then forward it to our
	// remote counterpart to be registered there.
	//
	// Conjugation is reversed because our proxy port will have the opposite
	// conjugation of the one registered.
	//
	// We don't need to create a proxy port unless we get a binding message from
	// our RCP.
	//
	
	{CriticalSectionGuard g(cs_);
		localRegistrations_[portName] = std::make_pair(contract->getTypeID(), !contract->isConjugate());
		BOOST_FOREACH(SimpleSocket::SocketPtr sock, clients_)
		{
			if( sock->valid() )
				this->sendMessage(sock, PRM_Register(portName, contract->getTypeID(), !contract->isConjugate()));
		}
	}
}


void PortRouter::unregisterPort(const char* portName)
{
	// Not implemented by DBM...
	{CriticalSectionGuard g(cs_);
	}
}

void PortRouter::notifyNoReply(int replyId)
{
	{CriticalSectionGuard g(cs_);
		// Need a port index to get socket ptr.
		//this->sendMessage(PRM_NoReply(replyId));
	}
}


/*********************
 * Protected methods *
 *********************/

bool PortRouter::handleConnect(SimpleSocket::SocketPtr client)
{
TRACE_ENTRY(PortRouter::handleConnect);

	DBGMSG(("PortRouter::handleConnect(%d)\n",SOCKET(*client)))
	{CriticalSectionGuard g(cs_);
		if (clients_.end() != clients_.find(client))
		{
			SocketSet::iterator i = pendingConnections_.find(client);
			if (pendingConnections_.end() == i)
			{
				DBGMSG(("  SendMessage(sock:%d,PRM_Connect(%d,%s))\n",SOCKET(*client),localPort_,localAddress_.data()))
				this->sendMessage(client, PRM_Connect(localPort_, localAddress_));
			}
			else
			{
				DBGMSG(("  client existed already. delete socket:%d\n", (SOCKET)(*client)))
				G_LOG_1_SOFTWARE("PortRouter.handleConnect pendingConnections_.erase(socket:%d)",(SOCKET)(**i));
				pendingConnections_.erase(i);
			}
		}
		else
		{
			DBGMSG(("  client not found. socket:%d\n", (SOCKET)(*client)))
			G_LOG_1_SOFTWARE("PortRouter.handleConnect client not found(socket:%d)",(SOCKET)(*client));
		}
	}
	// This is an error flag to the SocketServer base class.  Returning
	// false would exit the thread.  TBD:  Better error reporting???
	return true;
}


bool PortRouter::handleDisconnect(SimpleSocket::SocketPtr client)
{
	
	{CriticalSectionGuard g(cs_);
		// Close socket.
		DBGMSG(("PortRouter::handleDisconnect sock:%s", SOCKET(*client)))
		client->close(true);

		// Disconnect all our virtual ports.
		/*nextPortIndex_ = 0;
		portIndexContractMap_.clear();
		localEndPoints_.clearAll();
		remoteEndPoints_.clearAll();*/
	}
	return true;
}


bool PortRouter::handleMessage(SimpleSocket::SocketPtr client, const std::string& message)
{
	static const char* functionName = "PortRouter::handleMessage";
	
	TRACE_ENTRY(PortRouter::handleMessage);

	// First convert the bytes into a message.
	PortRouterMessage::Ptr msg = PortRouterMessage::Create(message);
	ASSERT(msg);

	if (bindAckWait_.wait(0))
	{
		if (msg->getId() == PortRouterMessage::RMID_BIND_ACK)
		{
			DBGMSG(("PortRouter::handle PRM_Bind_Ack\n"))
			bindAckWait_.reset();
			bindReply_ = ::boost::dynamic_pointer_cast<PRM_Bind_Ack>(msg);

			bindAckEvent_.set();

			// Handle any messages that were queued up while we were waiting for the BIND_ACK.
			//DBGMSG(("  Will pop up %d msg to dispach them.\n", todoList_.size()))
			while(!todoList_.empty())
			{
				G_LOG_2_TRACE("%s: todoList_...dispatch(%s)", functionName, toString(msg->getId()));
			//	msg = todoList_.front();
			//	DBGMSG(("    pop up %s.\n", toString(msg->getId())))
			//	if(PortRouterMessage::RMID_SEND == msg->getId()){
			//		PRM_Send *pMsg = dynamic_cast<PRM_Send*>(msg.get());
			//		DBGMSG(("    will dispatch(PRM_Send sender:%s,msgId:%d,replyId:%d)\n", pMsg->senderName().data(), pMsg->messageId(),pMsg->replyId()))
			//	}
				todoList_.front()->dispatch(this);
				todoList_.pop_front();
			}
		}
		else
		{
			//DBGMSG(("PortRouter::handleMessage(%d, %s) push it back\n", SOCKET(*client),toString(msg->getId())))
			// We're waiting for a BindAck, but this ain't it.  Queue it up for later dispatching.
			G_LOG_2_TRACE("%s: todoList_.push_back(%s)", functionName, toString(msg->getId()));
			todoList_.push_back(msg);
		}
	}
	else
	{
		replySock_ = client;

		//
		// Everyone's favorite:  Double-Dispatch!  WUPEE!
		//

		// register remote port for late SPI start
		if (PortRouterMessage::RMID_CONNECT == msg->getId())
		{
			BOOST_FOREACH(const PortNameContractMap::value_type& reg, localRegistrations_)
			{
				DBGMSG(("PortRouter::handleMessage  SendMessage(%d, PRM_Register(%s)//late SPI start\n",SOCKET(*client),reg.first.c_str()))
				this->sendMessage(client, PRM_Register(reg.first.c_str(), reg.second.first, reg.second.second));
			}
		}
		
		// Now we do the DD magic.  We pass the base class this, and it calls our overloaded
		// dispatch with its this so we get the derived class pointer WITHOUT a big switch to check
		// the type, downcast and call the appropriate method for each type.
		G_LOG_3_TRACE("%s: dispatch(%s, ID: %d)", functionName, toString(msg->getId()), msg->getId());
		msg->dispatch(this);

		replySock_.reset();
	}
	// Again, this is an error flag that would exit the thread, so we're always successful...TBD: ERROR HANDLING???
	return true;
}


IEndPointSP PortRouter::getEndPoint(PortIndex::PrimaryIndex portIndex)
{
	LocalEndPointSP retVal(new LocalEndPoint(this, portIndex));
	
	{CriticalSectionGuard g(cs_);
		// Create the endpoint and check for errors.
		size_t replication = remoteEndPoints_.add(portIndex, retVal);
		if (PortIndex::REPLICATION_BROADCAST != replication)
		{
			// The endpoint map returns the replication.  Update endpoint.
			retVal->setReplication(replication);
		}
		else
		{
			// TBD:  Extra cleanup???  Why would this ever fail???  Also, is binding prepared to handle
			// a null return from here?  Probably not.
			retVal.reset();
		}
	}
	return retVal;
}


bool PortRouter::setEndPoint(PortIndex& portIndex, IEndPointSP endPoint)
{
TRACE_ENTRY(PortRouter::setEndPoint);
	DBGMSG(("PortRouter::setEndPoint(%d, d)\n", portIndex))

	// These are the endpoints we use to to send messages to other objects.
	// This method is called by the binding ctor.
	//
	// This is just copied from active object.  Local endpoints point to local objects.
	// They get set by the binding ctor and we use them here just like ActiveObject.
	bool retVal = false;
	
	{CriticalSectionGuard g(cs_);
		localEndPoints_.resize(remoteEndPoints_.size());
		size_t replication = localEndPoints_.add(portIndex.port(), endPoint);
		if (PortIndex::REPLICATION_BROADCAST != replication)
		{
			portIndex.replication(replication);
			retVal = true;
		}
		// If this is a local binding request for a remote port, send it to RCP.
		// The question is whether this is due to a local binding.  If so, we need
		// to forward it to RCP.  We need to send it the name it registered with us
		// which we received when we handled the register message.
		PortRegistrationMap::const_iterator i = remoteRegistrations_.find(portIndex.port());
		if (remoteRegistrations_.end() != i)  // local bind
		{
			const RemoteRegistration& remoteRegistration = i->second;
			ASSERT(!remoteRegistration.name.empty());
			DBGMSG(("  SendMessage(%d,PRM_Bind(%s). local waiting...\n",SOCKET(*remoteRegistration.socket), remoteRegistration.name.data()))
			this->sendMessage(remoteRegistration.socket, PRM_Bind(remoteRegistration.name, portIndex));

			int timeout = -1;		// Wait forever for now....
			bindAckWait_.set();
			if (bindAckEvent_.wait(timeout))
			{
				// Map local port index to remote port index.  Since the bind was reqested locally,
				// the requestor has the binding, not us.  Our binding is just socket and remote index.
				RemoteBinding binding = {i->second.socket, bindReply_->portIndex()};
				remoteBindings_[portIndex] = binding;
			}
			else
			{
				DBGMSG(("  Synch binding message failed.  Now we're stuck!\n"))
				G_LOG_0_ERROR("Synch binding message failed.  Now we're stuck!");
				ASSERT_MSG(false, "Synch binding message failed.  Now we're stuck!");
			}
			// TODO:  ERROR HANDLING!
		}
		else // Otherwise use primary port index to pass port index back to binding handler.
		{    // remote bind
			// Save mapping
			DBGMSG(("  remote bind. save binding port %d\n", portIndex))
			remoteBindings_[portIndex.port()].portIndex = portIndex;
		}
	}
	return retVal;
}


bool PortRouter::clearEndPoint(const PortIndex& portIndex)
{
	// These are the endpoints we ues to to send messages to other objects.
	// This method is called by the binding dtor.
	//
	// Unlike the AO, we may have already cleared endpoints if RCP disconnected.
	bool retVal = true;
	
	{CriticalSectionGuard g(cs_);
		if ((portIndex.port() < localEndPoints_.size()) &&
			(portIndex.replication() < localEndPoints_.replicationSize(portIndex.port())))
		{
			retVal = localEndPoints_.clear(portIndex.port(), portIndex.replication());
		}
	}
	return retVal;
}


IContractCheckerSPC PortRouter::getContractChecker(PortIndex portIndex) const
{
	
	{CriticalSectionGuard g(cs_);
		PortIndexContractMap::const_iterator i = portIndexContractMap_.find(portIndex.port());
		ASSERT(portIndexContractMap_.end() != i)
		return i->second;
	}
}

#define ASSERT_BOUND(condition, message) \
	if (!(condition)) \
	throw SendMessageError(this->getObjectName(), \
							(message)->portIndex(), \
							(message)->messageID(), \
							(message)->releasePayload(), \
							SendMessageError::SMER_PORT_NOT_BOUND);

void PortRouter::put(RoomMessageSP message)
{
	
	{CriticalSectionGuard g(cs_);
		// Check if we're still connected.
		//ASSERT_BOUND(clients_, message);

		// Get contract type for this message.
		PortIndexContractMap::const_iterator iContract = portIndexContractMap_.find(message->portIndex().port());
		ASSERT_BOUND((portIndexContractMap_.end() != iContract), message);
		IContractCheckerSPC contract = iContract->second;
		ASSERT_BOUND(contract, message);
		ContractTypeId contractId = contract->getTypeID();

		// Get remote port index for this binding.
		PortIndexMap::const_iterator iBind = remoteBindings_.find(message->portIndex());
		ASSERT_BOUND((remoteBindings_.end() != iBind), message);
		const RemoteBinding& remoteBinding = iBind->second;

		// If this is a new message, we need to assign a reply id and store the reply endpoint.
		int replyId = 0;
		if (message->replyEndpoint())
		{
			ASSERT(!message->replyId());

			replyId = this->nextReplyId();
			ASSERT_BOUND((replyMap_.end() == replyMap_.find(replyId)), message);
			replyMap_[replyId] = message->replyEndpoint();
		}
		else // We assign the reply id from the message.
		{
			replyId = message->replyId();
		}
		DBGMSG(("PortRouter::put(sock:%d,PRM_Send sender:%s,msgId:%d %s, replyId:%d)\n", SOCKET(*remoteBinding.socket),message->senderName(), message->messageID(), contract->messageName(message->messageID()), replyId))
		G_LOG_5_SOFTWARE("PortRouter::put(sock:%d,PRM_Send sender:%s,msgId:%d %s, replyId:%d)", SOCKET(*remoteBinding.socket),message->senderName(), message->messageID(), contract->messageName(message->messageID()), replyId);
		this->sendMessage(
			remoteBinding.socket,
			PRM_Send(contractId, remoteBinding.portIndex, message->messageID(), message->releasePayload(), message->senderName(), replyId));
	}
}



void PortRouter::sendMessage(SimpleSocket::SocketPtr sock, const PortRouterMessage& msg)
{
	TRACE_ENTRY(PortRouter::sendMessage);
	ASSERT(sock);
	bool ok;
	if (sock)
	{
		// Using boost serialization.  We provide a string stream so we can get the string.
		std::ostringstream ost;
		oarchive oar(ost);

		// First this we add is always the message id so we can decode the message that follows.
		int id = msg.getId();
		oar << id << msg;
		ok = sock->sendMessage(ost.str());
		if(!ok) {
			DBGMSG(("Failed to send message to sock:%d, msgId:%d\n", SOCKET(*sock),id))
		}

		//DBGMSG(("PortRouter::sendMessage(d, %s)\n", toString(msg.getId())))
		G_LOG_1_SOFTWARE("PortRouter::sendMessage(%s)", toString((PortRouterMessage::RouterMessageId) id));
	}
}


template<typename REPLY_T>
bool PortRouter::sendMessage(SimpleSocket::SocketPtr sock, const PortRouterMessage& msg, REPLY_T& reply)
{
	TRACE_ENTRY(PortRouter::sendMessage__reply);

	bool retVal = false;

	ASSERT(sock);
	if (sock)
	{
		DBGMSG(("PortRouter::sendMessage(%d, %s, %d)\n", SOCKET(*sock),toString(msg.getId()),(PortRouterMessage::RouterMessageId)reply.getId()))
		this->sendMessage(sock, msg);
		retVal = true;
#pragma message(__HERE__"DualLane ToDo: implement alternative to synchronous send/receive messaging which interferes with receiveMessage loop in ActiveObject::main()->SocketServer::handleEvent()")
		std::string msgStr;
		if (sock->receiveMessage(msgStr) && !msgStr.empty())
		{
			std::istringstream iss(msgStr);
			iarchive iar(iss);
			PortRouterMessage::RouterMessageId msgId;
			iar >> msgId;
			iar >> reply;
			PortRouterMessage::RouterMessageId replyMsgId = (PortRouterMessage::RouterMessageId)reply.getId();
			retVal = (replyMsgId == msgId);
			if (!retVal)
				G_LOG_4_ERROR("PortRouter::sendMessage__reply: Expected %d[%s], Got %d[%s]", replyMsgId, toString(replyMsgId), msgId, toString(msgId));
		}
		else
		{
			DBGMSG(("!PortRouter::sendMessage(%d, %s, %d) Error:no(empty) reply\n", SOCKET(*sock),toString(msg.getId()),(PortRouterMessage::RouterMessageId)reply.getId()))
			G_LOG_0_ERROR("PortRouter::sendMessage__reply: no (or empty) reply!");
		}
	}else{
		DBGMSG(("!PortRouter::sendMessage(%d, %s, %d) Error\n", SOCKET(*sock),toString(msg.getId()),(PortRouterMessage::RouterMessageId)reply.getId()))
	}
	return retVal;
}

void PortRouter::dispatch(PRM_Connect* msg)
{
	TRACE_ENTRY("PortRouter::dispatch(PRM_Connect)");
	DBGMSG(("PortRouter::dispatch(PRM_Connect %s)\n", toString(msg->getId())))

	SimpleSocket::SocketPtr client = this->connect(msg->port(), msg->address().c_str());
	{CriticalSectionGuard g(cs_);
		// clients_.insert(client);   // Already done by the call to this->connect(...)
		pendingConnections_.insert(client);
	}
}

void PortRouter::dispatch(PRM_Register* msg)
{
	TRACE_ENTRY("PortRouter::dispatch(PRM_Register)");

	// This gets called when a remote dynamic port is registered.
	//
	// We create a virtual port and register it with DBM.  Now anyone locally can bind to it by name.
	// When we get a request for the endpoint as part of the binding process, we can send the binding
	// request to RCP to complete the remote binding.
	//
	PortIndex::PrimaryIndex virtualPort = PortIndex::INVALID;
	
	{CriticalSectionGuard g(cs_);
		SimpleSocket::SocketPtr sock = replySock_;
		// We need this guarded, but not the DBM call.
		virtualPort = nextPortIndex_++;

		// Create the endpoint and contract.
		remoteEndPoints_.resize(virtualPort + 1);
		portIndexContractMap_[virtualPort] = ProtocolFactory::Get(msg->typeId(), !msg->isConjugate());

		// Save mapping so we know remote port name to bind to.
		RemoteRegistration reg = {sock, msg->portName()};
		remoteRegistrations_[virtualPort] = reg;
		DBGMSG(("PortRouter::dispatch(sock:%d, PRM_Register:%s) port:%d\n", SOCKET(*sock),msg->portName().c_str(),msg->typeId(),virtualPort))
	}
	// Register a local port.
	DynamicBindingManager::Instance().registerDynamicPort(this, virtualPort, msg->portName().c_str(), true);
	// Todo:  prev returns success.  handle error.
}


void PortRouter::dispatch(PRM_Unregister* msg)
{
	TRACE_ENTRY("PortRouter::dispatch(PRM_Unregister)");
	DBGMSG(("PortRouter::dispatch(PRM_Unregister)\n"))
}


void PortRouter::dispatch(PRM_Bind* msg)
{
	TRACE_ENTRY("PortRouter::dispatch(PRM_Bind)");
	DBGMSG(("PortRouter::dispatch(PRM_Bind %s) sock:%d\n", msg->portName().c_str(), (SOCKET)(*replySock_)))
	SimpleSocket::SocketPtr sock = replySock_;

	// This gets called when a remote client binds to a port that was registered locally.
	// We need to create a virtual port and use DBM to bind it to the local object that registered it.
	//
	PortIndex::PrimaryIndex virtualPort = PortIndex::INVALID;
	
	{CriticalSectionGuard g(cs_);
		// We need this guarded, but not the DBM call.
		virtualPort = nextPortIndex_++;

		// Look up the contract info for the local port.
		ASSERT(!msg->portName().empty());
		PortNameContractMap::const_iterator i = localRegistrations_.find(msg->portName());
		ASSERT(localRegistrations_.end() != i); // TODO:  ERROR HANDLING!!!

		// Create the endpoint and contract.
		remoteEndPoints_.resize(virtualPort + 1);
		portIndexContractMap_[virtualPort] = ProtocolFactory::Get(i->second.first, i->second.second);

		// At this point, we are about to bind and that will call back to set endpoint.  It will use the
		// primary index of remoteBindings_ to pass back the local end point that was used.
	}
	// Bind to local port.
	BindingSP newBinding = DynamicBindingManager::Instance().dynamicBind(this, virtualPort, msg->portName().c_str());
	
	{CriticalSectionGuard g(cs_);
		// Now we can finish adding the remote binding.
		PortIndex localPort = remoteBindings_[virtualPort].portIndex;
		RemoteBinding bind = {sock, msg->portIndex(), newBinding};
		remoteBindings_[localPort] = bind;
		DBGMSG(("  SendMessage(sock:%d,PRM_Bind_Ack:%d) portName:%s\n", SOCKET(*sock), localPort, msg->portName().data()))
		this->sendMessage(sock, PRM_Bind_Ack(localPort));
	}
}


void PortRouter::dispatch(PRM_Bind_Ack* msg)
{
	// Already handled in PortRouter::handleMessage
	DBGMSG(("PortRouter::dispatch(PRM_Bind_Ack)\n"))
}


void PortRouter::dispatch(PRM_Unbind* msg)
{
	TRACE_ENTRY("PortRouter::dispatch(PRM_Unbind)");
	DBGMSG(("PortRouter::dispatch(PRM_Unbind)\n"))
}

void PortRouter::dispatch(PRM_Send* msg)
{
	TRACE_ENTRY("PortRouter::dispatch(PRM_Send)");
	IContractCheckerSPC contract = ProtocolFactory::Get(msg->contractId());
	DBGMSG(("PortRouter::dispatch(PRM_Send sender:%s,msgId:%d %s,replyId:%d)\n", msg->senderName().data(), msg->messageId(), contract->messageName(msg->messageId()),msg->replyId()))
	G_LOG_4_SOFTWARE("PortRouter::dispatch(PRM_Send sender:%s,msgId:%d %s,replyId:%d)", msg->senderName().data(), msg->messageId(), contract->messageName(msg->messageId()),msg->replyId());

	{CriticalSectionGuard g(cs_);
		// Get the local endpoint.
		PortIndex localPort = msg->portIndex();
		IEndPointSP localEndpoint;

		// Check the reply id.  If it's a reply we need to dig up the endpoint.
		IEndPointSP replyEndpoint;
		int replyId = msg->replyId();
		if (replyId < 0)
		{
			ReplyMap::iterator iReply = replyMap_.find(-replyId);
			ASSERT(replyMap_.end() != iReply);
			localEndpoint = iReply->second;
			replyMap_.erase(iReply);
		}
		else // Get the local endpoint and create our own reply endpoint.
		{
			localEndpoint = localEndPoints_.get(localPort.port(), localPort.replication());
			if (replyId > 0)
			{
				replyEndpoint.reset(new RemoteReplyEndpoint(this, localPort, -replyId));
			}
		}
		if (localEndpoint)	// May be NULL during tear-down.
			localEndpoint->acceptMessage(msg->messageId(), msg->releasePayload(), msg->senderName().c_str(), replyEndpoint);
	}
}


void PortRouter::dispatch(PRM_NoReply* msg)
{
	TRACE_ENTRY("PortRouter::dispatch(PRM_NoReply)");
	
	{CriticalSectionGuard g(cs_);
		int replyId = msg->replyId();
		if (replyId < 0)
		{
			replyMap_.erase(-replyId);
		}
	}
}

int PortRouter::nextReplyId()
{
	{CriticalSectionGuard g(cs_);
		static int nextId;  // OK because class is singleton.
		if (++nextId < 1)
		{
			nextId = 1;
		}
		return nextId;
	}
}

}
